     /*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.nmpp.handler;

import java.util.ArrayList;
import java.util.List;
import org.apache.log4j.Logger;
import cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv;
import cn.ac.iie.di.ban.nmpp.metastore.TableSe2DBRule;
import java.util.HashMap;

/**
 *
 * @author alexmu
 */
public abstract class SE2DWWorker implements Runnable {

    protected TableSe2DWHandler tableSe2DBHandler;
    TableSe2DBRule tableSe2DBRule = null;
    List<TableSe2DBRule.Column> tableColumnSet = null;
    int batchSize;
    int columnSize;
    List<HashMap<String, Object>> docRecordSet = new ArrayList();
    long lastUpdateTime = -1;
    int timeout = -1;
    protected volatile boolean stopped;
    protected String dwType;
    long totalLoadNum = 0;
    private static org.apache.log4j.Logger logger = Logger.getLogger(SE2DWWorker.class.getName());

    public SE2DWWorker(TableSe2DWHandler pTableSe2DBHandler, String pDWType) {
        tableSe2DBHandler = pTableSe2DBHandler;
        tableSe2DBRule = tableSe2DBHandler.tableSe2DBRule;
        tableColumnSet = tableSe2DBRule.getColumnSet();
        batchSize = tableSe2DBRule.getBatchSize();
        columnSize = tableColumnSet.size();
//        initPSQL();
        dwType = pDWType;
        stopped = false;
        timeout = (Integer) RuntimeEnv.getParam(RuntimeEnv.TIMEOUT);
    }

//    private void initPSQL() {
//        String sql = "INSERT INTO " + tableSe2DBRule.getTableName() + "(COLUMNS) VALUES(VALUES_MARK)";
//        String columns = "";
//        String values = "";
//        for (int i = 0; i < this.columnSize; i++) {
//            if (i == 0) {
//                columns = tableColumnSet.get(i).getColumnName();
//                values = "?";
//            } else {
//                columns = columns + "," + tableColumnSet.get(i).getColumnName();
//                values = values + ",?";
//            }
//        }
//        this.pSQL = sql.replace("COLUMNS", columns).replace("VALUES_MARK", values);
//        logger.info("insert sql for table " + tableSe2DBRule.getTableName() + " is " + this.pSQL);
//    }

    public void run() {
        TableSe2DBRule tableSe2DBRule = tableSe2DBHandler.getTableSe2DBRule();
        Thread.currentThread().setName("newmpp" + dwType + "dw_worker_" + tableSe2DBRule.getTableName() + "_" + System.nanoTime());
        logger.info("write to table " + tableSe2DBRule.getTableName() + " in " + dwType);
        long currentTime = -1;
        while (true) {
            if (this.stopped) {//怎么跳
                logger.info("prepare to exit...");
                loadData0();
                break;
            }

            List data = tableSe2DBHandler.getData();
            if (data == null) {
                try {
                    Thread.sleep((Integer) RuntimeEnv.getParam(RuntimeEnv.WAIT_TIME));
                } catch (Exception ex) {
                }
            } else {
                if (lastUpdateTime == -1) {
                    lastUpdateTime = System.currentTimeMillis();
                }
                loadData(data);
            }

            if (lastUpdateTime != -1) {
                currentTime = System.currentTimeMillis();
                if (currentTime - lastUpdateTime >= timeout) {
                    logger.info("time out and load data forcelly " + currentTime + ":" + lastUpdateTime);
                    loadData0();
                    lastUpdateTime = -1;
                    logger.info("time out process successfully " + lastUpdateTime);
                }
            }
        }
        logger.info(Thread.currentThread().getName() + " exited");
    }

    public void shutdown() {
        stopped = true;
    }

    public void loadData(List<HashMap<String, Object>> pRecordSet) {
        docRecordSet.addAll(pRecordSet);
        if (docRecordSet.size() >= batchSize) {
            loadData0();
            lastUpdateTime = -1;
            docRecordSet.clear();
        }
    }

    public abstract void init() throws Exception;

    public abstract void loadData0();
}
